/*********************************************************************************
 *      Copyright:  (C) 2021 lastbreath<2631336290@qq.com>
 *                  All rights reserved.
 *
 *       Filename:  sub.c
 *    Description:  This file 
 *                 
 *        Version:  1.0.0(09/30/2021)
 *         Author:  lastbreath <2631336290@qq.com>
 *      ChangeLog:  1, Release initial version on "09/30/2021 08:40:22 PM"
 *                 
 ********************************************************************************/

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include<pthread.h>

#include "mosquitto.h"
#include "mqtt_client.h"

#define HOST "localhost"
#define PORT 1883
#define KEEP_ALIVE 60
#define MSG_MAX_SIZE 512

static int running = 1;
static short isConnect = 0;
static struct mosquitto *mosq = NULL;

static char s_topic[128] = {0};

REGEISTER_SUBSCRIBE_HANDLER client_subscribe_message = NULL;

/* Return mqtt client with subscribe statue 
    zero is connect failed or one is connect success
*/
short client_subscribe_connect_status()
{
    return isConnect;
}



static void my_connect_callback(struct mosquitto *mosq, void *obj, int reason_code)
{
   	printf("subscribe: on_connect: %s\n", mosquitto_connack_string(reason_code));
	if(reason_code != 0){
		/* If the connection fails for any reason, we don't want to keep on
		 * retrying in this example, so disconnect. Without this, the client
		 * will attempt to reconnect. */
		mosquitto_disconnect(mosq);
	}

    int rc = mosquitto_subscribe(mosq, NULL, s_topic, 1);
	if(rc != MOSQ_ERR_SUCCESS){
		fprintf(stderr, "Error subscribing: %s\n", mosquitto_strerror(rc));
		/* We might as well disconnect if we were unable to subscribe */
		mosquitto_disconnect(mosq);
	}

    isConnect = 1;

}


void my_disconnect_callback(struct mosquitto *mosq,void *obj,int rc)
{
    isConnect = 0;
    printf("Call the function: my_disconnect_callback\n");
    running = 0;
}

void my_subscribe_callback(struct mosquitto *mosq,void *obj,int mid,int qos_count,const int *granted_qos)
{
    printf("Call the function: on_subscribe\n");
}

void my_message_callback(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
    printf("my topic:%s\n", s_topic);

    printf("Recieve a message of %s: %s\n.",(char *)msg->topic,(char *)msg->payload);
    
    if(0 == strcmp(msg->payload, "quit")) {
    
        mosquitto_disconnect(mosq);
    }

    if (client_subscribe_message)
    {
        client_subscribe_message(msg);
    }
}

static void* subscribe_enevt_loop(void* topic)
{

    memset(s_topic, 0, sizeof(s_topic));
    memcpy(s_topic, topic, strlen(topic));

    /* create background thread block */
    mosquitto_loop_forever(mosq, -1, 1);


    return NULL;
}

/* mqtt client subscribe topic */
unsigned short mqtt_client_subscribe(const char* topic)
{
    pthread_t tid;
    int res = pthread_create(&tid, NULL, subscribe_enevt_loop, topic);

    usleep(200 * 1000); /* slepp 200ms */

    return 0;
}

/* mqtt client subscribe initinal connect */
unsigned short client_subscribe_init_connect(const char* host, 
            const char* port, 
            const char* username, const char* password)
{
    int ret;

    ret = mosquitto_lib_init();

    mosq = mosquitto_new(NULL, true, NULL);
    if(mosq == NULL)
    {
        printf("New sub_test error!\n");
        mosquitto_lib_cleanup();
        return -1;
    }

    mosquitto_username_pw_set(mosq, username, password);
    mosquitto_connect_callback_set(mosq, my_connect_callback);
    mosquitto_disconnect_callback_set(mosq, my_disconnect_callback);
    mosquitto_subscribe_callback_set(mosq, my_subscribe_callback);
    mosquitto_message_callback_set(mosq, my_message_callback);

    ret = mosquitto_connect(mosq, host, atoi(port), KEEP_ALIVE);

    if(ret)
    {
        printf("Connect server error!\n");
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    printf("Start!\n");

    return 0;
}

/* release resource */
void client_subscribe_release_resource()
{
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();

    printf("mqtt client release subscribe resource!\n");
}

